package com.chenjj.bigdata.flink.stream;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * 通过DataStream API的方式统计socket输入的单词个数
 *
 * 用法e.g
 *
 * 1.在192.168.152.134上启动一个nc服务
 * nc -l 9000
 *
 * 2.启动本示例，或者将本示例导成jar包，在finlk服务上运行，host=192.168.152.134 port=9000
 *
 * 3.在192.168.152.134的nc上输出单词
 *
 * 4.本示例会统计期间你输入的单词个数
 */
public class SocketWordCount {


    static Logger logger  = LoggerFactory.getLogger(SocketWordCount.class);

    public static void main(String[] args) throws Exception {

        // get the execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // get input data by connecting to the socket
        DataStream<String> text = env.socketTextStream("192.168.152.135", 9000, "\n");

        env.enableCheckpointing(1000);
        env.setStateBackend(new FsStateBackend("file:///tmp/flink/checkpoints"));
        // parse the data, group it, window it, and aggregate the counts
        text.flatMap(new FlatMapFunction<String, WordWithCount>() {
                    @Override
                    public void flatMap(String value, Collector<WordWithCount> out) {
                        for (String word : value.split("\\s")) {
                            out.collect(new WordWithCount(word, 1L));
                        }
                    }
                })
                .keyBy("word")
                .sum("count")
                .print();



        env.execute("Socket WordCount");
    }

    // Data type for words with count
    public static class WordWithCount {

        public String word;
        public long count;

        public WordWithCount() {}

        public WordWithCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return word + " : " + count;
        }
    }
}